package cn.middle.remote.netty;

import java.io.File;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.middle.remote.Client;
import cn.middle.remote.Listener;
import cn.middle.remote.Node;
import cn.middle.remote.enums.NodeState;
import cn.middle.remote.enums.NodeType;
import cn.middle.remote.exception.RemoteException;
import cn.middle.remote.handler.client.ClientProcessingHandler;
import cn.middle.remote.handler.decodec.CodecAdapter;
import cn.middle.remote.handler.netty.UserHandler;
import cn.middle.remote.message.BaseSignal;
import cn.middle.remote.serialization.JsonSerialization;
import cn.middle.remote.serialization.Serialization;
import cn.middle.remote.util.OsUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;

public abstract class NettyClient<T extends Listener> extends Client implements cn.middle.remote.Channel,Node{

	private  final String threadWorkerName = getClass().getSimpleName() + "workThread";

	private final Logger log = LoggerFactory.getLogger(getClass());

	private EventLoopGroup nioEventLoopGroup;

	private volatile Channel channel;

	@SuppressWarnings("rawtypes")
	private ClientProcessingHandler handler;

	protected Listener listener;

	protected Serialization serialization;

	private volatile NodeState nodeState = NodeState.NOT_INITIALIZED;
	
	private Timer timer = new Timer();
	
	private TimerTask testConnectionTask ;
	
	/**
	 * 前置工作
	 */
	@SuppressWarnings("unchecked")
	private void front() {
		if(this.config == null) 
			throw new NullPointerException("the user config is null");
		if (listener == null)
			throw new NullPointerException("the userlistener is null");
		if (serialization == null)
			serialization = new JsonSerialization();
		
		handler = createProcessingHandler();
		handler.setConfig(config);
		handler.setListener(listener);
		handler.setSerialization(serialization);
	}
	
	
	@SuppressWarnings("rawtypes")
	protected abstract ClientProcessingHandler createProcessingHandler();
	
	public abstract void setListener(T listener);
	
	@Override
	public void start() {
		front();
		nodeState = NodeState.CONNECTIONS;
		Bootstrap bootstrap = new Bootstrap();
		if (OsUtil.isLinux()) {
			nioEventLoopGroup = new EpollEventLoopGroup(this.config.getThreadNum(),
					new DefaultThreadFactory(threadWorkerName, true));
		} else {
			nioEventLoopGroup = new NioEventLoopGroup(this.config.getThreadNum(),
					new DefaultThreadFactory(threadWorkerName, true));
		}
		bootstrap.group(nioEventLoopGroup).option(ChannelOption.SO_KEEPALIVE, true)
				.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
				.channel(NioSocketChannel.class);

		if (this.config.getConnectTimeout() < 3000) {
			bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
		} else {
			bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.config.getConnectTimeout());
		}

		bootstrap.handler(new ChannelInitializer<SocketChannel>() {

			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				CodecAdapter adapter = new CodecAdapter();
				sslContext(ch);
				ch.pipeline().addLast("ping", new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS))
						.addLast("decoder", adapter.getDecode()).addLast("encoder", adapter.getEncode())
						.addLast("handler", new UserHandler(handler));
			}
		});

		bootstrap.remoteAddress(config.getRemoteHost(), config.getRemotePort());
		ChannelFuture future = bootstrap.connect();
		boolean ret = future.awaitUninterruptibly(config.getConnectTimeout(), TimeUnit.MILLISECONDS);
		if (ret && future.isSuccess()) {
			this.channel = future.channel();
			nodeState = NodeState.CONNECTED;
		}
		//启动定时器检测重连
		timerConnected();
	}
	

	// 参考文档 https://www.cnblogs.com/codehello/articles/16924966.html
	private void sslContext(Channel ch) {
		if (config.isActiveSSL() && config.getSslConfig() != null) {
			try {
				// 引入SSL安全验证
				File certChainFile = new File(config.getSslConfig().getCertChainPath());
				File keyFile = new File(config.getSslConfig().getKeyPath());
				File rootFile = new File(config.getSslConfig().getCaPath());
				SslContext sslContext = SslContextBuilder.forClient().keyManager(certChainFile, keyFile)
						.trustManager(rootFile).build();
				ch.pipeline().addLast(sslContext.newHandler(ch.alloc()));
			} catch (Throwable t) {
				log.error(t.getMessage(), t);
			}
		}
	}



	private synchronized void reConnect() throws SocketException {
		stop(); // 先停止，后启动
		start();
	}

	public <B extends BaseSignal> void send(B signal)  throws RemoteException {
		
		if(!isNormal()) {
			//网络异常
			throw new RemoteException("Network connection exception");
		}
		channel.writeAndFlush(signal);
	}

	@Override
	public NodeType getNodeType() {
		return config.getUserType();
	}

	@Override
	public boolean isNormal() {
		return channel != null ? channel.isActive() : false;
	}

	@Override
	public boolean isClose() {
		return channel != null ? !channel.isOpen() : true;
	}

	@Override
	public NodeState getNodeState() {
		return nodeState;
	}

	@Override
	public void close() {
		nodeState = NodeState.CLOSING;
		if (nioEventLoopGroup != null) {
			nioEventLoopGroup.shutdownGracefully();
		}
		
		if(channel != null) {
			channel.close();
			nodeState = NodeState.CLOSE;
		}

	}

	@Override
	public void close(int timeout) {
		close();
	}

	@Override
	public SocketAddress getRemoteSocketAddress() {
		return channel != null ? channel.remoteAddress() : null;
	}

	@Override
	public SocketAddress getLocalSocketAddress() {
		return channel != null ? channel.localAddress() : null;
	}


	@Override
	public void serialization(Serialization serialization) {
		this.serialization = serialization;
	}


	@Override
	public void excepton(Throwable t) {
		log.error(t.getMessage(),t);
		
	}


	@Override
	public void stop() {
		this.close();
		
	}

	private void testingConnected() {

		if (!isNormal()) {
			try {
				log.info("reConnect start");
				reConnect();
				log.info("reConnect end");
			} catch (Exception e) {
				log.error(e.getMessage(),e);
			}
		}
	}
	
	/**
	 * 定时执行重连任务
	 */
	private synchronized void timerConnected() {
		
		if(testConnectionTask != null) {
			testConnectionTask.cancel();
			//timer.cancel();
		}
		
		testConnectionTask = new TimerTask() {
			
			@Override
			public void run() {
				testingConnected() ;
			}
		};
		
		int periodTime = this.config.getConnectTimeout() + 100;
		if(periodTime < 3000){
			periodTime = 3000 + 100;
		}
		timer.scheduleAtFixedRate(testConnectionTask, periodTime, periodTime);		
	}

}
